package sync;

import common.Const;
import configer.SqlServerConfiger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rocketmq.Producer;
import utils.Druid;
import utils.PrintResultSetUtil;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

public class SqlServerSync {

    protected final static Logger logger = LoggerFactory.getLogger(SqlServerSync.class);     // 日志对象

    private SqlServerConfiger sqlServerConfiger;                                             // 配置对象
    private Map<String, String> batchPage;

    public SqlServerSync(SqlServerConfiger sqlServerConfiger, Map<String, String> batchPage) {
        logger.info("实例化 <SqlServer数据库全量同步> 处理对象");
        this.sqlServerConfiger = sqlServerConfiger;
        this.batchPage = batchPage;
    }

    /**
     * @Descriotion 预测试程序能否正常运行
     */
    public void testRun(Druid druid) {
        // 当测试不成功时，Flag++
        int Flag = -1;
        String sql = " USE " + sqlServerConfiger.getSqlServer_database()
                + ";" + "SELECT TOP 1 * FROM "
                + sqlServerConfiger.getSqlServer_tableName() + ";";
        do {
            logger.info("全量同步程序 执行前测试 ～");
            logger.info("数据库连接池 初始化");
            Connection conn = druid.getConnection();
            if (conn == null) {
                logger.info("全量同步程序 执行前测试不通过，无法继续进行。");
                druid.close(conn);
                try {
                    Thread.sleep(3000);
                    Flag++;
                    continue;
                } catch (InterruptedException e) {
                    logger.error("Thread.sleep 失败 { " + e + " } ");
                }
            }

            Statement stmt = druid.getStmt(conn);
            if (stmt == null) {
                logger.info("全量同步程序 执行前测试不通过，无法继续进行。");
                druid.close(stmt, conn);
                try {
                    Thread.sleep(3000);
                    Flag++;
                    continue;
                } catch (InterruptedException e) {
                    logger.error("Thread.sleep 失败 { " + e + " } ");
                }
            }

            try {
                ResultSet rs = stmt.executeQuery(sql);
                logger.info("获取 ResultSet <" + rs + "> 成功");
                PrintResultSetUtil.Print(rs);
                druid.close(rs, stmt, conn);
                Flag = 1000;
            } catch (SQLException throwables) {
                logger.error("获取 ResultSet 出错 { " + throwables + " } ");
                logger.info("全量同步程序 执行前测试不通过，无法继续进行。");
                try {
                    Thread.sleep(3000);
                    Flag++;
                    druid.close(stmt, conn);
                    continue;
                } catch (InterruptedException e) {
                    logger.error("Thread.sleep 失败 { " + e + " } ");
                }
            }

        } while (Flag < 0);

        if (Flag != 1000) {
            System.exit(0);
        }
    }


    public void run(Druid druid) {
        boolean flag = true;
        logger.info("全量同步程序 开始执行 ～");
        // 初始化 indexFiledValue
        initSqlServerPage();
        // 获取一个批次的数据
        while (flag) {
            flag = getBatchData(druid);
        }

    }


    /**
     * @Descriotion 初始化分页的页数
     */
    private void initSqlServerPage() {
        if (batchPage.get(sqlServerConfiger.getSqlServer_database()
                + "." + sqlServerConfiger.getSqlServer_schema()
                + "." + sqlServerConfiger.getSqlServer_tableName()) == null) {
            synchronized (batchPage) {
                batchPage.put(sqlServerConfiger.getSqlServer_database()
                        + "." + sqlServerConfiger.getSqlServer_schema()
                        + "." + sqlServerConfiger.getSqlServer_tableName(), "1");
            }
        }
    }


    /**
     * @Descriotion 获取SqlServer全量数据
     */
    private boolean getBatchData(Druid druid) {
        boolean flag = true;
        Connection conn = druid.getConnection();
        Statement stmt = druid.getStmt(conn);

        int page = Integer.parseInt(batchPage.get(sqlServerConfiger.getSqlServer_database()
                + "." + sqlServerConfiger.getSqlServer_schema()
                + "." + sqlServerConfiger.getSqlServer_tableName()));
        String sql = "USE " + sqlServerConfiger.getSqlServer_database() + ";"
                + "SELECT TOP " + Const.PAGE_SIZE + " * FROM ("
                + " SELECT ROW_NUMBER() OVER (ORDER BY "
                + sqlServerConfiger.getDataIndexs()
                + ") AS RowNumber,* FROM "
                + sqlServerConfiger.getSqlServer_tableName()
                + ") AS A WHERE RowNumber > "
                + Const.PAGE_SIZE * (page - 1)
                + " ;";
        logger.info("SQL语句为 " + sql);
        ResultSet rs = null;
        try {
            rs = stmt.executeQuery(sql);
            if (rs.next()) {
                // 初始化 Producer 对象
                Producer producer = new Producer(sqlServerConfiger.getProducerGroup(), sqlServerConfiger.getNameSrv());
                // 遍历 rs 发送到 RocketMQ
                producer.sqlServerProduceSync(rs, sqlServerConfiger, sqlServerConfiger.getTopicName(), sqlServerConfiger.getTagName());
                producer.close();
                synchronized (batchPage) {
                    batchPage.put(sqlServerConfiger.getSqlServer_database() + "." + sqlServerConfiger.getSqlServer_schema() + "."
                            + sqlServerConfiger.getSqlServer_tableName(), String.valueOf(page + 1));
                }
            } else {
                flag = false;
            }
        } catch (SQLException throwables) {
            logger.error("获取 ResultSet 失败 " + throwables);
        } finally {
            druid.close(rs, stmt, conn);
        }

        return flag;

    }

}

